package rabitmq

import (
	"errors"
	"fmt"
	"gitee.com/lv_baobao/gcore"
	"gitee.com/lv_baobao/gcore/logHelper"
	"github.com/streadway/amqp"
	"sync"
	"time"
)

type RabbitMqConnPool struct {
	pool                   chan *amqp.Connection
	maxPoolSize            int
	currentConnectionIndex int32
	lock                   sync.Mutex
	closeChan              chan bool
}

var rabbitMqConnPool *RabbitMqConnPool
var rabbitMqConnPoolOnce sync.Once

func NewRabbitMqConnPool() *RabbitMqConnPool {
	rabbitMqConnPoolOnce.Do(func() {
		rabbitMqConnPool = new(RabbitMqConnPool)
		poolSize := gcore.NewAppSettingsHelper().GetAppConfig().RabbitMq.PoolCount
		if poolSize == 0 {
			poolSize = 20
		}
		rabbitMqConnPool.pool = make(chan *amqp.Connection, poolSize)
		rabbitMqConnPool.maxPoolSize = poolSize
		rabbitMqConnPool.preInitConnPool()
		rabbitMqConnPool.lock = sync.Mutex{}
		rabbitMqConnPool.closeChan = make(chan bool)
		go func() {
			for range rabbitMqConnPool.closeChan {
				fmt.Println("收到rabbitmq连接池重新初始化请求")
				time.Sleep(time.Second * 2)
				for true {
					err := rabbitMqConnPool.preInitConnPool()
					if err == nil {
						break
					}
				}
			}
		}()
	})
	return rabbitMqConnPool
}

//预热
func (p *RabbitMqConnPool) preInitConnPool() error {
	logHelper.Info(fmt.Sprintf("开始预热"))
	for i := 0; i < p.maxPoolSize; i++ {
		conn := p.createConn()
		if conn != nil {
			p.pool <- conn
		} else {
			time.Sleep(time.Second)
		}
	}
	if len(p.pool) == 0 {
		fmt.Println(fmt.Sprintf("预热rabbitmq连接数%d失败", p.maxPoolSize))
		p.clearPool()
		return errors.New("预热rabbitmq连接失败")
	} else {
		logHelper.Info(fmt.Sprintf("预热rabbitmq连接数%d成功", p.maxPoolSize))
	}

	return nil
}

//clearPool
func (p *RabbitMqConnPool) clearPool() {
	p.pool = nil
}

//createConn
func (p *RabbitMqConnPool) createConn() *amqp.Connection {
	config := gcore.NewAppSettingsHelper().GetAppConfig().RabbitMq
	connectUrl := fmt.Sprintf("amqp://%s:%s@%s:%d/%s", config.UserName, config.PassWord, config.Host, config.Port, config.VirtualHost)
	conn, err := amqp.Dial(connectUrl)
	if err != nil {
		logHelper.Error("创建rabbitmq connection失败", err)
		return nil
	}
	return conn
}

//func (p *RabbitMqConnPool) GetFreeRabbitMqConn(timeout time.Duration) (*amqp.Connection, error) {
//	if p.pool == nil {
//		p.lock.Lock()
//		defer p.lock.Unlock()
//		if p.pool == nil {
//			p.closeChan <- true
//		}
//	}
//	ticker := time.NewTicker(timeout)
//	select {
//	case _ = <-ticker.C:
//		return nil, errors.New("pool is empty,timeout")
//	case connection, ok := <-p.pool:
//		if !ok {
//			return nil, errors.New("pool is empty,pool is closed")
//		}
//		if connection.IsClosed() {
//			if len(p.pool) == 0 {
//				p.clearPool()
//			}
//			return nil, errors.New("connection isClosed")
//		}
//		return connection, nil
//	}
//}

func (p *RabbitMqConnPool) GetFreeRabbitMqConn(timeout time.Duration) (*amqp.Connection, error) {
	if p.pool == nil {
		p.lock.Lock()
		defer p.lock.Unlock()
		if p.pool == nil {
			p.closeChan <- true
		}
	}
	connection, ok := <-p.pool
	if !ok {
		return nil, errors.New("pool is empty,pool is closed")
	}
	if connection.IsClosed() {
		if len(p.pool) == 0 {
			p.clearPool()
		}
		return nil, errors.New("connection isClosed")
	}
	return connection, nil

}

func (p *RabbitMqConnPool) Free(conn *amqp.Connection) error {
	if conn == nil {
		logHelper.Error("free rabbitmq connection error, connection is nil")
		return nil
	}
	p.pool <- conn
	return nil
}
